package com.guass.nettyprotostuff.rpc.server.handler;

import com.guass.nettyprotostuff.disruptor.RingBufferWorkerPoolFactory;
import com.guass.nettyprotostuff.rpc.common.CommonHandler;
import com.guass.nettyprotostuff.rpc.common.ModelMsg;
import com.guass.nettyprotostuff.rpc.common.TypeData;
import com.guass.nettyprotostuff.service.RpcService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

/**
 * @author guass
 */
@Slf4j
public class NodesChannelHandler extends CommonHandler {

    private RpcService rpcService;

    public NodesChannelHandler(RpcService rpcService) {
        this.rpcService = rpcService;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ModelMsg hotKeyMsg) throws Exception {
        handlerData(channelHandlerContext,hotKeyMsg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        if (rpcService != null) {
            rpcService.onTerminalActive(ctx);
        }
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("guass exceptionCaught {}, {}， {}",ctx.channel().id().asLongText(),rpcService, cause.getMessage());
        if (rpcService != null) {
            rpcService.onTerminalInActive(ctx);
        }
        ctx.close();
    }

    /**
     * channel不可写时关闭autoRead
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        if (!ctx.channel().isWritable()) {
            Channel channel = ctx.channel();
            channel.config().setAutoRead(false);
        }
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        if (channel.isWritable()) {
            channel.config().setAutoRead(true);
        }
    }

    @Override
    protected void handlerData(ChannelHandlerContext ctx, ModelMsg msg) {
        if (msg.getMessageType() == TypeData.PING) {
            sendPongMsg(ctx);
        }
        else {
            RingBufferWorkerPoolFactory.getInstance().getMessageProducer("guass").onData(ctx,msg);
        }

    }

    @Override
    protected void handlerReaderIdle(ChannelHandlerContext ctx) {
        // TODO Auto-generated method stub
        super.handlerReaderIdle(ctx);
        ctx.close();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (rpcService != null) {
            rpcService.onTerminalInActive(ctx);
        }
        ctx.close();
    }
}
